package org.codehaus.activemq.web;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.HashMap;
import java.util.Map;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.servlet.ServletContext;
import javax.servlet.http.HttpSession;
import javax.servlet.http.HttpSessionActivationListener;
import javax.servlet.http.HttpSessionEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.ActiveMQConnection;
import org.codehaus.activemq.ActiveMQConnectionFactory;
import org.codehaus.activemq.ActiveMQSession;

public class WebClient
  implements HttpSessionActivationListener, Externalizable
{
  public static final String webClientAttribute = "org.codehaus.activemq.webclient";
  public static final String connectionFactoryAttribute = "org.codehaus.activemq.connectionFactory";
  public static final String brokerUrlInitParam = "org.codehaus.activemq.brokerURL";
  public static final String embeddedBrokerInitParam = "org.codehaus.activemq.embeddedBroker";
  private static final Log log = LogFactory.getLog(WebClient.class);
  private transient ServletContext context;
  private static transient ConnectionFactory factory;
  private transient ActiveMQConnection connection;
  private transient ActiveMQSession session;
  private transient MessageProducer producer;
  private transient Map consumers = new HashMap();

  public static WebClient getWebClient(HttpSession session)
  {
    return (WebClient)session.getAttribute("org.codehaus.activemq.webclient");
  }

  public WebClient()
  {
  }

  public WebClient(ServletContext context)
  {
    this.context = context;
    setupConnectionFactory(context);
  }

  public void start() throws JMSException {
  }

  public void stop() throws JMSException {
    try {
      this.connection.close();
    }
    finally {
      this.producer = null;
      this.session = null;
      this.connection = null;
      this.consumers.clear();
    }
  }

  public void writeExternal(ObjectOutput out) throws IOException {
  }

  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    this.consumers = new HashMap();
  }

  public void send(Destination destination, Message message) throws JMSException {
    if (this.producer == null) {
      this.producer = getSession().createProducer(null);
    }
    log.info("Sending to destination: " + destination);
    this.producer.send(destination, message);
    log.info("Sent! message: " + message);
  }

  public Session getSession() throws JMSException {
    if (this.session == null) {
      this.session = ((ActiveMQSession)getConnection().createSession(false, 1));
    }
    return this.session;
  }

  public ActiveMQConnection getConnection() throws JMSException {
    if (this.connection == null) {
      this.connection = ((ActiveMQConnection)factory.createConnection());
      this.connection.start();
    }
    return this.connection;
  }

  public void sessionWillPassivate(HttpSessionEvent event) {
    try {
      stop();
    }
    catch (JMSException e) {
      log.warn("Could not close connection: " + e, e);
    }
  }

  public void sessionDidActivate(HttpSessionEvent event)
  {
    this.context = event.getSession().getServletContext();
    setupConnectionFactory(this.context);
  }

  private static synchronized void setupConnectionFactory(ServletContext context) {
    factory = initConnectionFactory(context);
    if (factory == null) {
      log.warn("No ConnectionFactory available in the ServletContext for: org.codehaus.activemq.connectionFactory");
      factory = new ActiveMQConnectionFactory("vm://localhost");
      context.setAttribute("org.codehaus.activemq.connectionFactory", factory);
    }
  }

  public static ConnectionFactory initConnectionFactory(ServletContext servletContext) {
    ConnectionFactory connectionFactory = (ConnectionFactory)servletContext.getAttribute("org.codehaus.activemq.connectionFactory");
    if (connectionFactory == null) {
      String brokerURL = servletContext.getInitParameter("org.codehaus.activemq.brokerURL");

      servletContext.log("Value of: org.codehaus.activemq.brokerURL is: " + brokerURL);

      if (brokerURL == null) {
        brokerURL = "vm://localhost";
      }

      boolean embeddedBroker = MessageServletSupport.asBoolean(servletContext.getInitParameter("org.codehaus.activemq.embeddedBroker"));
      servletContext.log("Use embedded broker: " + embeddedBroker);

      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);
      factory.setUseEmbeddedBroker(embeddedBroker);

      connectionFactory = factory;
      servletContext.setAttribute("org.codehaus.activemq.connectionFactory", connectionFactory);
    }
    return connectionFactory;
  }

  public synchronized MessageConsumer getConsumer(Destination destination) throws JMSException {
    MessageConsumer consumer = (MessageConsumer)this.consumers.get(destination);
    if (consumer == null) {
      consumer = this.session.createConsumer(destination);
      this.consumers.put(destination, consumer);
    }
    return consumer;
  }
}